package com.cy.juc;

import com.cy.util.U;
import com.google.common.collect.Lists;
import org.testng.annotations.Test;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;

public class CompletableFuture_BSite_Base_Learn {

	public static final String filter_words_text = "filter_words.text";
	public static final String news_text = "news.text";

	// 1.Future vs CompletableFuture
	// 1.2 Future的局限性
	@Test(description = "原始的实现方式")
	public void t1_2() throws ExecutionException, InterruptedException {
		ExecutorService executor = Executors.newFixedThreadPool(3);
		// 1.读取敏感词汇
		Future<String[]> filterWords_future = executor.submit(() -> U.readResourcesFileOriginal("filter_words.text").split(","));
		// 2.读取新闻稿
		Future<String> news_future = executor.submit(() -> U.readResourcesFileOriginal("news.text"));
		// 3.替换
		Future<String> replace_future = executor.submit(() -> {
			String[] filterWords = filterWords_future.get();
			String news = news_future.get();
			System.out.println("原始的新闻稿:" + news);
			for (String filterWord : filterWords) {
				if (news.contains(filterWord)) {
					news = news.replace(filterWord, "**");
				}
			}
			return news;
		});
		System.out.println("最终的新闻稿:" + replace_future.get());
	}
	// 上面的Future问题
	// 在没有阻塞的情况下，无法对Future的结果执行进一步的操作。Future不会告知你它什么时候完成，你如果想要得到结果，必须通过一个get()方法，该方法会阻塞直到结果可用为止。 它不具备将回调函数附加到Future后并在Future的结果可用时自动调用回调的能力。
	// 无法解决任务相互依赖的问题。filterWordFuture和newsFuture的结果不能自动发送给replaceFuture，需要在replaceFuture中手动获取，所以使用Future不能轻而易举地创建异步工作流。
	// 不能将多个Future合并在一起。假设你有多种不同的Future，你想在它们全部并行完成后然后再运行某个函数，Future很难独立完成这一需要。
	// 没有异常处理。Future提供的方法中没有专门的API应对异常处理，还是需要开发者自己手动异常处理


	// 1.3 CompletableFuture 的优势
	// CompletableFuture 实现了Future和CompletionStage接口
	// CompletableFuture 相对于 Future 具有以下优势:
	// 		。为快速创建、链接依赖和组合多个Future提供了大量的便利方法。
	// 		。提供了适用于各种开发场景的回调函数，它还提供了非常全面的异常处理支持
	// 无缝衔接和亲和lambda 表达式和 Stream - API。
	// 我见过的真正意义上的异步编程，把异步编程和函数式编程、响应式编程多种高阶编程思维集于一身，设计上更优雅。


	// 2创建异步任务
	// 2.1 runAsync
	@Test
	public void t2_1() {
		U.printThreadLog("main start");
		CompletableFuture.runAsync(() -> {
			U.printThreadLog("读取文件开始");
			U.sleep(1);
			U.printThreadLog("读取文件结束");
		});
		U.printThreadLog("here are not blocked,main continue");
		U.sleep(3);
		U.printThreadLog("mian end");
	}

	// 疑问: 异步任务是并发执行还是并行执行?
	// 如果是单核CPU，那么异常任务之间就是并发执行，如果是多核CPU(多CPU) 步任务就是并行执行
	// 重点(敲黑板)水
	// 作为开发者，我们只需要清楚如何开启异步任务，CPU硬件会把异步任务合理的分配给CPU上的核运行水

	// 2.2 supplyAsync
	// completableFuture.runAsync( 开启不带返回结果异步任务。但是，如果您想从后台的异步任务中返回一个结果怎么办?此时completableFuture.supp1yAsync是你最好的选择了。
	@Test
	public void t2_2() throws ExecutionException, InterruptedException {
		U.printThreadLog("main start");
		CompletableFuture<String> news_future = CompletableFuture.supplyAsync(() -> {
			U.printThreadLog("开始读取任务");
			return U.readResourcesFileOriginal("news.text");
		});
		U.printThreadLog("here not blocked main continue");
		System.out.println(news_future.get());
	}

	// 2.3异步任务中的线程池
	// 小知识点  通过ForkJoinPool.commonPool管理的线程池 归main线程管理 main结束了 他也就结束了 所以程序跑完 不会亮红灯
	// 但是如果是自定义线程池 不归main管理 所以结束了还是会亮红灯 在testNG中无法观察到此现象 需要使用main方法执行
	@Test
	public void t2_3() throws ExecutionException, InterruptedException {
		ExecutorService executor = Executors.newFixedThreadPool(5);
		U.printThreadLog("main start");
		CompletableFuture<String> news_future = CompletableFuture.supplyAsync(() -> {
			U.printThreadLog("开始读取任务");
			return U.readResourcesFileOriginal("news.text");
		}, executor);
		U.printThreadLog("here not blocked main continue");
		System.out.println(news_future.get());
		executor.shutdown();
	}

	// 2.4 异步编程思想
	// 综合上述，看到了吧，我们没有显式地创建线程，更没有涉及线程通信的概念，整个过程根本就没涉及线程知识吧，以上专业的说法是: 线程的创建和线程负责的任务进行解耦，它给我们带来的好处线程的创建和启动全部交给线程池负责，具体任务的编写就交给程序员，专人专事。
	// 异步编程是可以让程序并行(也可能是并发运行的一种手段，其可以让程序中的一个工作单元作为异步任务与主线程分开独立运行，并且在异步任务运行结束后，会通知主线程它的运行结果或者失败原因，毫无疑问，一个异步任务其实就是开启一个线程来完成的，使用异步编程可以提高应用程序的性能和响应能力等
	// 		作为开发者，只需要有一个意识:
	// 		开发者只需要把耗时的操作交给CompletableFuture开启一个异步任务，然后继续关注主线程业务，当异步任务运行完成时会通知主线程它的运行结果。我们把具备了这种编程思想的开发称为异步编程思想。

	// 3.异步任务回调
	// CompletableFuture,get()方法是阻塞的。调用时它会阻塞等待 直到这个Future完成，并在完成后返回结果。但是，很多时候这不是我们想要的。
	// 对于构建异步系统，我们应该能够将回调附加到CompletableFuture上，当这个Future完成时，该回调应自动被调用。这样，我们就不以等待结里了，然后在Future的回调函数内编写完成Future之后需要执行的逻辑，您可以使用thenApply0，thenAccept0)和thenRun0 方法，它们可以把回调函数附加到CompletableFuture

	// 3.1 thenApply 对异步任务的结果 进一步  使用Function 转换
	@Test
	public void t3_1() throws ExecutionException, InterruptedException {
		U.printThreadLog("main start");
		CompletableFuture<String> readFile_future = CompletableFuture.supplyAsync(() -> {
			U.printThreadLog("读取文件");
			return U.readResourcesFileOriginal("filter_words.text");
		});

		CompletableFuture<String[]> filterWords_future = readFile_future.thenApply(x -> x.split(","));
		U.printThreadLog("main continue");
		System.out.println("filterWords" + Arrays.toString(filterWords_future.get()));
		U.printThreadLog("main end");
	}

	// 3.2 thenAccept  对异步结果进行消费使用
	// 如果你不想从回调函数返回结果，而只想在Future完成后运行一些代码，则可以使用thenAccept)
	// 这些方法是入参一个Consumer<T>，它可以对异步任务的执行结果进行消费使用，方法返回CompletableFuture<Void>。
	@Test
	public void t3_2() {
		CompletableFuture.supplyAsync(() -> U.readResourcesFileOriginal("filter_words.text"))
				.thenApply(words -> words.split(","))
				.thenAccept(x -> U.printThreadLog(Arrays.toString(x))
				);
		U.sleep(2);
	}

	// 3.3 thenRun
	// 前面我们已经知道，通过thenApply(Function<T,R>)对链式操作中的上一个异步任务的结果进行转换，返回一个新的结果
	// 通过thenAccept(Consumer<T>)对链式操作中上一个异步任务的结果进行消费使用，不返回新结果
	// 如果我们只是想从CompletableFuture的链式操作得到一个完成的通知，甚至都不使用上一步链式操作的结果，那么CompletableFuture.thenRun) 会是你最佳的选择，
	// 它需要一个Runnable并返回completableFuture<Void>。 一般用于在末端使用
	@Test
	public void t3_3() {
		CompletableFuture.supplyAsync(() -> U.readResourcesFileOriginal("filter_words.text")).thenRun(() -> U.printThreadLog("filter_words.text 读取完毕"));
		U.sleep(1);
	}

	// 3.4 更异步提升并行化
	// CompletableFuture 提供的所有回调方法都有两个异步变体
	// CompletabTeFuture<U> thenApp1y(Function<T,U> fn)
	// // 回调方法的异步变体(异步回调)
	// CompTetabTeFuture<U> thenApp1yAsync(Function<T,U> fn)
	// CompletabeFuture<u> thenApplyAsync(Function<T,U> fn， Executor executor)
	// 注意: 这些带了Async的异步回调 通过在单独的线程中执行回调任务 来帮助您进一步促进并行化计算。
	// 回顾需求:异步读取 flter_words.txt 文件中的内容，读取完成后，转换成敏感词数组，主线程获取结果打印输出这个数组

	// 前言
	// 一般而言，commonPool为了提高性能*
	// 		*thenApply中回调任务和suppLyAsync中的异步任务使用的是同一个线程
	//  *特殊情况:
	// 		*如supplyAsync中的任务是立即返回结果( 不是耗时的任务 ),thenApply回调任务也会在主线程执行

	@Test
	public void t3_4_0() throws ExecutionException, InterruptedException {
		U.printThreadLog(
				Arrays.toString(
						CompletableFuture.supplyAsync(() -> {
							U.printThreadLog("读取文件");
							return U.readResourcesFileOriginal("filter_words.text");
						}).thenApply(filterWords -> {
							U.printThreadLog("转换文件内容为数组");
							return filterWords.split(",");
						}).get()
				)
		);
	}

	// 这里可能会不太好观察到 thenApplyAsync (异步回调) 的效果 需要多运行几次
	// 但是只要理解为 thenApplyAsync 会尽可能使用线程池内的其他线程执行即可
	@Test
	public void t3_4_1() throws ExecutionException, InterruptedException {
		U.printThreadLog(
				Arrays.toString(
						CompletableFuture.supplyAsync(() -> {
							U.printThreadLog("读取文件");
							return U.readResourcesFileOriginal("filter_words.text");
						}).thenApplyAsync(filterWords -> {
							U.printThreadLog("转换文件内容为数组");
							return filterWords.split(",");
						}).get()
				)
		);
	}

	// 此外，如果将Executor传递给 thenApplyAsync 回调，则该回调的异步任务将在从Exeutor的线程池中获取的线程中执行
	// 注意:这些带了Async的异步回调**通过在单独的线程中执行回调任务** 来帮助您进一步促进并行化计算.

	// 其他两个回调的变体版本如下:
	// // thenAccept和其异步回调
	// CompletableFuture<Void> thenAccept(Consumer<T> action)
	// CompletableFuture<Void>thenAcceptAsync(Consumer<T> action)
	// CompletableFuture<Void> thenAcceptAsync(Consumer<T> action， Executor executor)
	// // thenRun和其异步回调
	// CompletableFuture<Void> thenRun(Runnable action)
	// CompletableFuture<Void> thenRunAsync(Runnable action)
	// CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor)
	@Test
	public void t3_4_2() throws ExecutionException, InterruptedException {
		ExecutorService executor = Executors.newFixedThreadPool(4);
		U.printThreadLog(
				Arrays.toString(
						CompletableFuture.supplyAsync(() -> {
							U.printThreadLog("读取文件");
							return U.readResourcesFileOriginal("filter_words.text");
						}).thenApplyAsync(filterWords -> {
							U.printThreadLog("转换文件内容为数组");
							return filterWords.split(",");
						}, executor).get()
				)
		);
	}

	// 4 异步任务编排
	// 4.1 编排2个依赖关系的异步任务 thenCompose()
	// *thenApply(Function<T,R>)
	// 重心在于对上一步异步任务的结果T进行应用转换，经Function回调转换后的结果R是一个简单的值*
	// *thenCompose( Function<T， CompletableFuture<R> >)
	// 重心在于对上一步异步任务的结果T进行应用，经Function回调转换后的结果是一个CompletableFuture对象*
	// 经验:
	// *编排两个依赖关系的异步任务( CompletableFuture 对 ) 请使用 thenCompose() 方法
	@Test
	public void t4_1() throws ExecutionException, InterruptedException {
		System.out.println(
				Arrays.toString(
						CompletableFuture.supplyAsync(() -> {
							U.printThreadLog("读取文件");
							return U.readResourcesFileOriginal(filter_words_text);
						}).thenCompose(
								filter_words_text -> {
									U.printThreadLog("分割文件");
									return CompletableFuture.supplyAsync(() -> filter_words_text.split(","));
								}
						).get()
				));
	}

	// 4.2 thenCombine() 编排2个非依赖关系的异步分任务
	// 我们已经知道，当其中一个Future依题于另一个Future，使用 thencompose) 用合两Future。如里两Future间没有依赖关系，你希望两个Future独立运行并在两者都完成之后执行回调操作时，则使用 thencombine()
	// T是第一个任务的结果 U是第二个任务的结果 V是经BiFunction应用转换后的结果
	// CompletableFuture<V> thenCombine(CompletableFuture<u> other, BiFunction<T,U,V> func)

	// thenCombine / thenAcceptBoth / runAfterBoth都表示：将两个CompletableFuture组合起来，只有这两个都正常执行完了，才会执行某个任务。
	// 区别在于：
	// thenCombine：会将两个任务的执行结果作为方法入参，传递到指定方法中，且有返回值
	// thenAcceptBoth: 会将两个任务的执行结果作为方法入参，传递到指定方法中，且无返回值
	// runAfterBoth 不会把执行结果当做方法入参，且没有返回值。
	@Test
	public void t4_2() throws ExecutionException, InterruptedException {
		CompletableFuture<String[]> future1 = CompletableFuture.supplyAsync(() -> {
			U.printThreadLog("读取敏感词文件");
			return U.readResourcesFileOriginal(filter_words_text).split(",");
		});
		CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
			U.printThreadLog("读取新闻稿");
			return U.readResourcesFileOriginal(news_text);
		});
		System.out.println(future1.thenCombine(future2, (filterWords, news) -> {
			for (String filterWord : filterWords) {
				if (news.contains(filterWord)) {
					news = news.replace(filterWord, "**");
				}
			}
			return news;
		}).get());
	}

	// 4.3 合并多个异步任务  allOf / anyOf
	// allOf
	// completableFuture.a110f 用于以下情形中: 有多个需要立并行运行的Future，并在所有这些 Future 都完成后执行一些操作.
	// 需求: 判断文件中是否包含sb
	@Test
	public void t4_3_0() {
		List<String> tests = Lists.newArrayList(filter_words_text, filter_words_text, news_text);

		List<CompletableFuture<Boolean>> contain_sb_futures = tests.stream().map(
				x -> CompletableFuture.supplyAsync(() -> U.readResourcesFileOriginal(x).contains("sb")
				)
		).collect(Collectors.toList());

		System.out.println(
				CompletableFuture.allOf(contain_sb_futures.toArray(new CompletableFuture[tests.size()]))
						.thenApply(aVoid -> contain_sb_futures.stream().map(CompletableFuture::join).filter(x -> x).count())
						.join()
		);
	}

	// anyOf
	// 顾名思义，当给定的多个异步任务中的有任意Future一个完成时，需要执行一些操作，可以使用 anyof 方法 但是返回值是Object需要注意
	@Test
	public void t4_3_1() throws ExecutionException, InterruptedException {
		CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
			U.sleep(3);
			return "future1完成了";
		});
		CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
			U.sleep(1);
			return "future2完成了";
		});
		CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
			U.sleep(2);
			return "future3完成了";
		});
		CompletableFuture<Object> resFuture = CompletableFuture.anyOf(future1, future2, future3);
		System.out.println(resFuture.get());
	}

	// 5.异步任务的异常处理
	// 如果在 supplyAsync 任务中出现异常，后续的 thenApply 和 thenAccept 回调都不会执行，
	// CompletableFuture 将转入异常处理如果在第一个thenApply任务中出现异常，第二个thenApply和 最后的 thenAccept 回调不会被执行，completableFuture 将转入异常处理，依次类推。
	@Test
	public void t5_0() {
		CompletableFuture.supplyAsync(() -> "result1" + 1 / 0)
				.thenApply(in -> in + "result2")
				.thenAccept(U::printThreadLog);
	}

	// 5.1 exceptionally()
	// exceptionally 用于处理回调链上的异常，回调链上出现的任何异常，回调链不继续向下执行，都在exceptionally中处理异常。
	@Test
	public void t5_1() {
		System.out.println(CompletableFuture.supplyAsync(() -> "result1" + 1 / 0)
				.thenApply(in -> in + "result2")
				.exceptionally(new Function<Throwable, String>() {
					@Override
					public String apply(Throwable throwable) {
						throwable.printStackTrace();
						return "有异常了";
					}
				}).join());
	}

	// 5.2 handle()
	// CompletableFuture API还提供了一种更通用的方法 handle( 表示从异常中恢复handle() 常常被用来恢复回调链中的一次特定的异常，回调链恢复后可以进一步向下传递
	// 异步任务不管是否发生异常，handle方法都会执行所以，handle核心作用在于对上一步异步任务进行现场修复 如果发生异常，则res参数将为null，否则ex参数将为null
	@Test
	public void t5_2() {
		System.out.println(
				CompletableFuture.supplyAsync(() -> {
					int a = 1 / 0;
					return "result1";
				}).handle((s, ex) -> {
					U.printThreadLog("进入handle1了");
					if (ex != null) {
						return "发生了异常1 原因是 " + ex.getMessage();
					}
					return s;
				}).thenApply(input -> {
					int a = 1 / 0;
					return input + " result2";
				}).handle((s, ex) -> {
					U.printThreadLog("进入handle2了");
					if (ex != null) {
						return "发生了异常2 原因是 " + ex.getMessage();
					}
					return s;
				}).join()
		);
	}
	// 和以往一样，为了提升并行化，异常处理可以方法单独的线程执行，以下是它们的异步回调版本
	// CompletableFuture<R> exceptiona1y(Function<Throwable，R> fn)
	// CompletableFuture<R> exceptional1yAsync(Function<Throwable，R> fn)  jdk17+
	// CompletableFuture<R> exceptionallyAsync(Function<Throwable, R> fn,Executor executor)  idk17+
	// CompletableFuture<R> handle(BiFunction<T,Throwable,R> fn)
	// CompletableFuture<R> handleAsync(BiFunction<T,Throwable,R> fn)
	// CompletableFuture<R> handleAsync(BiFunction<T,Throwable,R> fn, Executor executor)

	@Test
	public void t0() {
		CompletableFuture<Integer> future = CompletableFuture.completedFuture(1);
		System.out.println(future.join());
	}


	// whenComplete
	// 表示，某个任务执行完成后，执行的回调方法，无返回值；并且whenComplete方法返回的CompletableFuture的result是上个任务的结果。
	@Test
	public void t0_1() {
		CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
			U.sleep(TimeUnit.SECONDS, 3);
			return "执行完完毕";
		});

		CompletableFuture<String> f2 = f1.whenComplete((s, throwable) -> {
			System.out.println(s);
			System.out.println(throwable);
		});

		System.out.println(f2.join());
	}

	// 其他文章 https://zhuanlan.zhihu.com/p/622768247
	public void t0_2(){

	}
}
